package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.storage.StorageLevel;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SA7601;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.DateUtil;

public class DomainAnalyzTask_SA7601ZT_city_dayNew implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA7601ZT_city_dayNew.class);
	private static final long serialVersionUID = 1L;
	private static String outName;
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		long temp=start;
		long temp_end=start;
		double exectime=0.0;
		
		System.out.println("==============DomainAnalyzTask_SA7601ZT_city_dayNew begin====================");
		System.out.println("............本省主体(地市)....开始............");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try {
			String p_linces_key = new String(ConfigUtil.get("LICENSE_KEY").getBytes("ISO-8859-1"),"UTF-8");
			System.out.println("LICENSE_KEY="+p_linces_key);
			DataFrame dataSource = dataFrames.get("day_domain");
			String ztsql = "select top_domain,sum(visit_count)count from day_domain_temp where license = 1 group by top_domain";
			DataFrame dfzt_sa7601=dataSource.sqlContext().sql(ztsql);
			System.out.println("............生成day_domain临时表............");
		    dfzt_sa7601.registerTempTable("df_dc_temp");
		    dfzt_sa7601.persist(StorageLevel.MEMORY_AND_DISK_SER());
	        
	        String sql = "select a.site_name,a.license_key,a.top_domain,c.shengid,c.shiid "
	        		+ "from br2005_temp a, ICP_GJCX_BAXX_YMLB_temp b,ICP_GJCX_BAXX_ZT_temp c where a.top_domain=b.ym and b.ztid =c.ztid ";
	        DataFrame df_zt = dataSource.sqlContext().sql(sql);
	        System.out.println("............生成br2005zt临时表............");
	        df_zt.registerTempTable("br2005zt_temp");
	        df_zt.persist(StorageLevel.MEMORY_AND_DISK_SER());
	        
	        System.out.println("............topdomain 合并(域名在br2005中)............");
	        String sql_beian = "select * from ("
	        		+ "select a.top_domain,b.shiid,a.count,"
	        		+ "(CASE WHEN ISNULL(b.site_name) = false THEN b.site_name ELSE  'N/A' END) AS site_name, "
	        		+ "(CASE WHEN ISNULL(b.license_key) = false THEN b.license_key ELSE  'N/A' END) AS license_key, "
					+ "ROW_NUMBER() over(partition by b.shiid order by a.count desc)rank "
					+ "from df_dc_temp a left join br2005zt_temp b on a.top_domain = b.top_domain where b.license_key like '%"+p_linces_key+"%' "
					+ "group by a.top_domain,site_name,license_key,b.shiid,a.count)s "
					+ "where s.rank <= 200"; 
	        temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
            
            String dayStr = commandMap.get("-d");
		    final Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
		    
	        DataFrame oper_tmp_beian = dataSource.sqlContext().sql(sql_beian);
	        
	        List<Row> list_row = oper_tmp_beian.toJavaRDD().take(6000);
			List<SA7601> list_SA7601 = new ArrayList<SA7601>();	
			if(list_row.size() > 0){
	        	for(Row row_top_domain : list_row){
	        	    SA7601 sa = new SA7601();
	        		sa.setYear(DateUtil.getCurrentYear(date));
	        		sa.setHalf_year(DateUtil.getHalfYear(date));
	        		sa.setQuarter(DateUtil.getQuarter(date));
	        		sa.setMonth(DateUtil.getCurrentMonth(date));
	        		sa.setWeek(DateUtil.getCurrentWeek(date));
	        		sa.setDay(DateUtil.getCurrentDay(date));
	        		sa.setBuss_type("WEBSITE_RANK_SUM_CITY");
	        		sa.setBuss_value(row_top_domain.getAs("shiid").toString());
	        		sa.setRank(Integer.parseInt(row_top_domain.getAs("rank").toString()));
	        		sa.setDomain(row_top_domain.getAs("top_domain").toString());
	        	    sa.setWebsite_name(row_top_domain.getAs("site_name").toString());
	        	    sa.setIcp_num(row_top_domain.getAs("license_key").toString());
	        	    sa.setTraffic(row_top_domain.getAs("count").toString());
	        		sa.setSta_range(2); //本省主体是2
	        		list_SA7601.add(sa);
	        	}
	        }
	        
	        
	        /*JavaRDD<SA7601> result = oper_tmp_beian.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>, SA7601>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Iterable<SA7601> call(Iterator<Row> t) throws Exception {
					List<SA7601> list_SA7601 = new ArrayList<SA7601>();
					while(t.hasNext()){
						SA7601 sa = new SA7601();
						Row row = t.next();
						//long count = Long.parseLong(String.valueOf(map.get("count")));
						int rank = Integer.parseInt(String.valueOf(row.getAs("rank")));
						sa.setYear(DateUtil.getCurrentYear(date));
						sa.setHalf_year(DateUtil.getHalfYear(date));
						sa.setQuarter(DateUtil.getQuarter(date));
						sa.setMonth(DateUtil.getCurrentMonth(date));
						sa.setWeek(DateUtil.getCurrentWeek(date));
						sa.setDay(DateUtil.getCurrentDay(date));
						sa.setBuss_type("WEBSITE_RANK_SUM_CITY");
						sa.setBuss_value(row.getAs("shiid").toString());
						sa.setRank(rank);
		        		sa.setDomain(row.getAs("top_domain").toString());
		        	    sa.setWebsite_name(row.getAs("site_name").toString());
		        	    sa.setIcp_num(row.getAs("license_key").toString());
		        	    sa.setTraffic(row.getAs("count").toString());
		        		sa.setSta_range(2); //本省主体是2
		        		list_SA7601.add(sa);
					}
					
					return list_SA7601;
				}
			});*/
	        
			dfzt_sa7601.sqlContext().dropTempTable("df_dc_temp");
			dfzt_sa7601.unpersist();
	        df_zt.sqlContext().dropTempTable("br2005zt_temp");
	        df_zt.unpersist();
	        
	        JavaSparkContext ctx = new JavaSparkContext(dataSource.sqlContext().sparkContext());
	    	JavaRDD<SA7601> rdd = ctx.parallelize(list_SA7601);
	    	DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SA7601.class);
			outList.add(new OutQueueEntity(outName, df));
		
			CommonUtils.deleteTaskTableData("SA7601", date, "WEBSITE_RANK_SUM_CITY", 2, "day");
		} catch (Exception e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("DomainAnalyzTask_SA7601ZT_New本省主体(地市):exectime: "+min+" min............");
		return outList;
	}

}














